package operator.transfer;

import java.util.Arrays;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import util.Printer;
import util.ThreadHelper;

class SwitchMapOperator {

    public static void main(String[] args) {
//        switchMap1();
        switchMap2();
    }

    static void switchMap1() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4);

        Observable.fromIterable(list)
                .switchMap(new Function<Integer, ObservableSource<String>>() {

                    @Override
                    public ObservableSource<String> apply(Integer integer) throws Exception {
                        return Observable.just("integer = " + integer);
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Printer.print(s);
                Printer.printThreadName("switchMap:");
            }
        });
    }

    static void switchMap2() {
        List<Integer> list = Arrays.asList(1, 2, 3, 4);

        Observable.fromIterable(list)
                .switchMap(new Function<Integer, ObservableSource<String>>() {

                    @Override
                    public ObservableSource<String> apply(Integer integer) throws Exception {
//                        return Observable.just("integer = " + integer).subscribeOn(Schedulers.newThread());
                        if (integer == 4) {
                            return Observable.create(new ObservableOnSubscribe<String>() {
                                @Override
                                public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
                                    observableEmitter.onNext("123 " + integer);
                                    observableEmitter.onComplete();
                                }
                            }).subscribeOn(Schedulers.io()).doOnComplete(new Action() {
                                @Override
                                public void run() throws Exception {
                                    Printer.print(integer + " doOnComplete");
                                }
                            }).doOnDispose(new Action() {
                                @Override
                                public void run() throws Exception {
                                    Printer.print(integer + " doOnDispose");
                                }
                            });
                        } else {
                            return Observable.just("integer = " + integer).subscribeOn(Schedulers.io()).doOnComplete(new Action() {
                                @Override
                                public void run() throws Exception {
                                    Printer.print(integer + " doOnComplete");
                                }
                            }).doOnDispose(new Action() {
                                @Override
                                public void run() throws Exception {
                                    Printer.print(integer + " doOnDispose");
                                }
                            });
                        }

                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Printer.print(s);
                Printer.printThreadName("switchMap:");
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {

            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Printer.printThreadName("switchMap onComplete");
            }
        });
        ThreadHelper.sleep();
    }


}
